package com.hxgis.common.service.impl;

import com.es.common.EsAggCondition;
import com.es.common.domain.EsCondition;
import com.es.common.domain.agg.Group;
import com.es.common.enums.FlagConnect;
import com.es.common.enums.Order;
import com.es.wrapper.DataWrapper;
import com.hxgis.common.base.ResultVO;
import com.hxgis.common.base.StationCache;
import com.hxgis.common.dto.ExtremeAnaStationDTO;
import com.hxgis.common.dto.ExtremeYearDataDTO;
import com.hxgis.common.dto.ProcInfoAnalysDTO;
import com.hxgis.common.entity.*;
import com.hxgis.common.repository.ExtremeProcRepository;
import com.hxgis.common.repository.ProcRainInfoRepository;
import com.hxgis.common.repository.StationInfoRepository;
import com.hxgis.common.service.ExtremeProcService;
import com.hxgis.common.service.RainDayService;
import com.hxgis.common.service.StationExtremeService;
import com.hxgis.common.service.StationService;
import com.hxgis.common.utils.NumUtil;
import com.hxgis.common.utils.RectField;
import com.hxgis.common.utils.ResultVOUtil;
import com.hxgis.common.vo.ExtremeAnalysisVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class ExtremeProcServiceImpl implements ExtremeProcService {

    @Autowired
    ExtremeProcRepository extremeProcRepository;

    @Autowired
    ProcRainInfoRepository procRainInfoRepository;

    @Autowired
    private StationExtremeService stationExtremeService;

    @Autowired
    private StationService stationService;

    @Autowired
    private StationInfoRepository stationInfoRepository;

    @Autowired
    RainDayService rainDayService;


    @Override
    public ResultVO queryExtremProcInfos(String startTime, String endTime, String leftTime, String rightTime, String flag, Integer type, String stationNum, String riverName) {
        Long st = System.currentTimeMillis();
        //查站点信息
        EsCondition stationCondition = EsCondition.getEsCondition();
        if (Objects.nonNull(riverName)){
            stationCondition.eq("river",riverName);
        }
        List<Station> stationList = stationInfoRepository.selectList(stationCondition);
        List<String> stationNoList = RectField.getBeanFields(stationList, "stationNo");



        EsAggCondition esAggCondition = EsAggCondition.getInstance();
        Group stationNoGroup = esAggCondition.Group("stationNo");
        stationNoGroup.size(10000);
        Group idGroup = stationNoGroup.groupSub("id").minExtFields("startTime","endTime","year");
        if (Objects.equals(type,1) && Objects.equals(flag,"20")){
            idGroup.min("totalRainFall_20");
        }else if (Objects.equals(type,1) && Objects.equals(flag,"08")){
            idGroup.min("totalRainFall_08");
        }else if (Objects.equals(type,2) && Objects.equals(flag,"20")){
            idGroup.min("continueDay_20");
        }else if (Objects.equals(type,2) && Objects.equals(flag,"08")){
            idGroup.min("continueDay_08");
        }else if (Objects.equals(type,3) && Objects.equals(flag,"20")){
            idGroup.min("notContinueDay_20");
        }else if (Objects.equals(type,3) && Objects.equals(flag,"08")){
            idGroup.min("notContinueDay_08");
        }

        idGroup.size(10000);
        EsCondition condition = esAggCondition.getCondition();
        condition.in("stationNo", com.hxgis.common.utils.CollectionUtils.convertCollectionString(stationNoList));
        if (Objects.equals(startTime.substring(0,4),endTime.substring(0,4))){
            condition.gte("endDays",startTime.substring(4,8)).lte(endTime.substring(4,8)).gte("startTime",leftTime).lte(rightTime);
        }else{
            condition.gte("startTime",leftTime).lte(rightTime);
            EsCondition ed = EsCondition.getEsCondition().gte("endDays", startTime.substring(4, 8)).lte("1231")
                    .or().gte("endDays", "0101").lte(endTime.substring(4, 8)).or();
            condition.subSearch(ed, FlagConnect.AND);
        }



        if (Objects.equals(type,1) && Objects.equals(flag,"20")){
            condition.eq("totalRainFall_20",null).not();
        }else if (Objects.equals(type,1) && Objects.equals(flag,"08")){
            condition.eq("totalRainFall_08",null).not();
        }else if (Objects.equals(type,2) && Objects.equals(flag,"20")){
            condition.eq("continueDay_20",null).not();
        }else if (Objects.equals(type,2) && Objects.equals(flag,"08")){
            condition.eq("continueDay_08",null).not();
        }else if (Objects.equals(type,3) && Objects.equals(flag,"20")){
            condition.eq("notContinueDay_20",null).not();
        }else if (Objects.equals(type,3) && Objects.equals(flag,"08")){
            condition.eq("notContinueDay_08",null).not();
        }
        if (Objects.nonNull(stationNum)){
            condition.eq("stationNo",stationNum);
        }
        List<ProcInfoAnalysDTO> procInfoAnalysDTOS = extremeProcRepository.selectListAgg(esAggCondition, new DataWrapper<ProcInfoAnalysDTO>() {
            @Override
            public ProcInfoAnalysDTO wrapper(Map<String, Object> data) {
                ProcInfoAnalysDTO procInfoAnalysDTO = new ProcInfoAnalysDTO();
                procInfoAnalysDTO.setStationNo(data.get("stationNo").toString());
                procInfoAnalysDTO.setStartTime(data.get("startTime").toString());
                procInfoAnalysDTO.setEndTime(data.get("endTime").toString());
                if (Objects.equals(type,1) && Objects.equals(flag,"20")){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("totalRainFall_20").toString())/10);
                }else if (Objects.equals(type,1) && Objects.equals(flag,"08")){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("totalRainFall_08").toString())/10);
                }else if (Objects.equals(type,2) && Objects.equals(flag,"20")){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("continueDay_20").toString()));
                }else if (Objects.equals(type,2) && Objects.equals(flag,"08")){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("continueDay_08").toString()));
                }else if (Objects.equals(type,3) && Objects.equals(flag,"20")){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("notContinueDay_20").toString()));
                }else if (Objects.equals(type,3) && Objects.equals(flag,"08")){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("notContinueDay_08").toString()));
                }
                procInfoAnalysDTO.setYear(data.get("year").toString());
                return procInfoAnalysDTO;
            }
        });
        log.info("极端事件查询es用时,{}",System.currentTimeMillis() - st);
        Map<String, List<ProcInfoAnalysDTO>> extremeProcInfoMaps = procInfoAnalysDTOS.stream().collect(Collectors.groupingBy(ProcInfoAnalysDTO::getStationNo));

        //查最大值
        Map<String, List<ProcInfoAnalysDTO>> extremeMaxInfo = getExtremeMaxInfo(stationNum, type, flag, leftTime, rightTime);

        //todo 站点阈值信息
        List<StationExtremeInfo> allExtreInfo = stationExtremeService.getAllExtreInfo();
        Map<String, List<StationExtremeInfo>> extremeInfoMaps = allExtreInfo.stream().collect(Collectors.groupingBy(StationExtremeInfo::getStationNo));

        //todo 单站相关信息
        Map<String, Station> stringStationMap = stationService.queryAllStation();
        List<ExtremeAnaStationDTO> extremeAnaStationDTOS = new ArrayList<>();
        if (Objects.nonNull(stationNum)){
            if (CollectionUtils.isEmpty(extremeProcInfoMaps)){
                ExtremeAnaStationDTO extremeAnaStationDTO = new ExtremeAnaStationDTO();
                extremeAnaStationDTO.setStationNo(stationNum);
                extremeAnaStationDTO.setStationName(stringStationMap.get(stationNum).getCity());
                extremeAnaStationDTO.setProvince(stringStationMap.get(stationNum).getProvince());
                extremeAnaStationDTO.setRiver(stringStationMap.get(stationNum).getRiver());
                extremeAnaStationDTO.setRegion(stringStationMap.get(stationNum).getRegion());
                extremeAnaStationDTO.setLat((Double.valueOf(stringStationMap.get(stationNum).getLat())));
                extremeAnaStationDTO.setLon((Double.valueOf(stringStationMap.get(stationNum).getLon())));
                extremeAnaStationDTO.setYear("999999");
                extremeAnaStationDTO.setObserDate("999999");
                extremeAnaStationDTO.setPreObserDate("999999");
                extremeAnaStationDTO.setValue(999999d);
                //查看阈值
                Double extremeValue = getExtremeProcValue(stationNum, type, flag, extremeInfoMaps);
                extremeAnaStationDTO.setExtremeValue(extremeValue);

                if (CollectionUtils.isEmpty(extremeMaxInfo.get(stationNum))){
                    extremeAnaStationDTO.setMaxObserDate("999999");
                    extremeAnaStationDTO.setMaxValue(999999d);
                }else{
                    Double value = Double.valueOf(extremeMaxInfo.get(stationNum).get(0).getValues());
                    extremeAnaStationDTO.setMaxObserDate(extremeMaxInfo.get(stationNum).get(0).getStartTime() + "-" + extremeMaxInfo.get(stationNum).get(0).getEndTime());
                    extremeAnaStationDTO.setMaxValue(value);
                }
                extremeAnaStationDTO.setIsExtreme(false);
                extremeAnaStationDTO.setIsMaxValue(false);
                extremeAnaStationDTOS.add(extremeAnaStationDTO);
                return ResultVOUtil.success(extremeAnaStationDTOS);
            }
            extremeProcInfoMaps.forEach((stationNo,procExtremeDTOList) -> {
                procExtremeDTOList.stream().forEach(x -> {
                    ExtremeAnaStationDTO extremeAnaStationDTO = new ExtremeAnaStationDTO();
                    extremeAnaStationDTO.setStationNo(stationNo);
                    extremeAnaStationDTO.setLat((Double.valueOf(stringStationMap.get(stationNo).getLat())));
                    extremeAnaStationDTO.setLon((Double.valueOf(stringStationMap.get(stationNo).getLon())));
                    extremeAnaStationDTO.setStationName(stringStationMap.get(stationNo).getCity());
                    extremeAnaStationDTO.setProvince(stringStationMap.get(stationNo).getProvince());
                    extremeAnaStationDTO.setRiver(stringStationMap.get(stationNo).getRiver());
                    extremeAnaStationDTO.setRegion(stringStationMap.get(stationNo).getRegion());
                    extremeAnaStationDTO.setYear(x.getYear());
                    extremeAnaStationDTO.setObserDate(x.getEndTime());
                    extremeAnaStationDTO.setPreObserDate(x.getStartTime());
                    extremeAnaStationDTO.setValue(x.getValues());
                    //查看阈值
                    Double extremeValue = getExtremeProcValue(stationNo, type, flag, extremeInfoMaps);
                    extremeAnaStationDTO.setExtremeValue(extremeValue);

                    if (CollectionUtils.isEmpty(extremeMaxInfo.get(stationNo))){
                        extremeAnaStationDTO.setMaxValue(999999d);
                        extremeAnaStationDTO.setMaxObserDate("999999");
                        extremeAnaStationDTO.setIsExtreme(false);
                        extremeAnaStationDTO.setIsMaxValue(false);
                    }else{
                        Double value = Double.valueOf(extremeMaxInfo.get(stationNo).get(0).getValues());
                        extremeAnaStationDTO.setMaxValue(value);
                        extremeAnaStationDTO.setMaxObserDate(extremeMaxInfo.get(stationNo).get(0).getStartTime() + "-" + extremeMaxInfo.get(stationNo).get(0).getEndTime());
                        extremeAnaStationDTO.setIsExtreme(x.getValues().compareTo(extremeValue) >= 0);
                        extremeAnaStationDTO.setIsMaxValue(x.getValues().compareTo(value) >= 0);
                    }
                    extremeAnaStationDTOS.add(extremeAnaStationDTO);
                });
            });
            log.info("极端事件 极端过程接口总用时{}",System.currentTimeMillis() - st );
            return ResultVOUtil.success(extremeAnaStationDTOS);
        }


        ExtremeAnalysisVO extremeAnalysisVO = new ExtremeAnalysisVO();
        //左侧地图数据信息 循环所有站点信息
        List<ExtremeAnaStationDTO> extremeAnaStationMapDTOS = new ArrayList<>();
        Map<String, List<ProcInfoAnalysDTO>> extremeInfoStationMaps = procInfoAnalysDTOS.stream().collect(Collectors.groupingBy(ProcInfoAnalysDTO::getStationNo));
//        List<Station> stations = StationCache.getStations();
        stationList.stream().forEach(station -> {
            List<ProcInfoAnalysDTO> anaStationDTOS = extremeInfoStationMaps.get(station.getStationNo());
            if (CollectionUtils.isEmpty(anaStationDTOS)){
                ExtremeAnaStationDTO extremeAnaStationDTO = getExtremeAnaStationDTO(flag, type, extremeMaxInfo, extremeInfoMaps, stringStationMap, station);
                extremeAnaStationMapDTOS.add(extremeAnaStationDTO);
            }else{
                //再将数据按照年分组查询当前年
                Map<String, List<ProcInfoAnalysDTO>> yearExtreMaps = anaStationDTOS.stream().collect(Collectors.groupingBy(ProcInfoAnalysDTO::getYear));
                //当前年信息
                List<ProcInfoAnalysDTO> extremeAnaStationDTOS1 = yearExtreMaps.get(startTime.substring(0, 4));
                if (CollectionUtils.isEmpty(extremeAnaStationDTOS1)){
                    ExtremeAnaStationDTO extremeAnaStationDTO = getExtremeAnaStationDTO(flag, type, extremeMaxInfo, extremeInfoMaps, stringStationMap, station);
                    extremeAnaStationMapDTOS.add(extremeAnaStationDTO);
                }else{
                    ExtremeAnaStationDTO extremeAnaStationDTO = new ExtremeAnaStationDTO();
                    extremeAnaStationDTO.setStationNo(station.getStationNo());
                    extremeAnaStationDTO.setStationName(stringStationMap.get(station.getStationNo()).getCity());
                    extremeAnaStationDTO.setProvince(stringStationMap.get(station.getStationNo()).getProvince());
                    extremeAnaStationDTO.setRiver(stringStationMap.get(station.getStationNo()).getRiver());
                    extremeAnaStationDTO.setRegion(stringStationMap.get(station.getStationNo()).getRegion());
                    extremeAnaStationDTO.setLat((Double.valueOf(stringStationMap.get(station.getStationNo()).getLat())));
                    extremeAnaStationDTO.setLon((Double.valueOf(stringStationMap.get(station.getStationNo()).getLon())));
                    extremeAnaStationDTO.setYear(startTime.substring(0,4));
                    extremeAnaStationDTO.setObserDate(extremeAnaStationDTOS1.get(0).getEndTime());
                    extremeAnaStationDTO.setPreObserDate(extremeAnaStationDTOS1.get(0).getStartTime());
                    Double nowVlaue = extremeAnaStationDTOS1.get(0).getValues();
                    extremeAnaStationDTO.setValue(nowVlaue);
                    //查看阈值
                    Double extremeValue = getExtremeProcValue(station.getStationNo(), type, flag, extremeInfoMaps);
                    extremeAnaStationDTO.setExtremeValue(extremeValue);

                    if (CollectionUtils.isEmpty(extremeMaxInfo.get(station.getStationNo()))){
                        extremeAnaStationDTO.setMaxObserDate("999999");
                        extremeAnaStationDTO.setMaxValue(999999d);
                        extremeAnaStationDTO.setIsExtreme(false);
                        extremeAnaStationDTO.setIsMaxValue(false);
                    }else{
                        Double value = Double.valueOf(extremeMaxInfo.get(station.getStationNo()).get(0).getValues());
                        extremeAnaStationDTO.setMaxValue(value);
                        extremeAnaStationDTO.setIsExtreme(nowVlaue.compareTo(extremeValue) >= 0);
                        extremeAnaStationDTO.setIsMaxValue(nowVlaue.compareTo(value) >= 0);
                        extremeAnaStationDTO.setMaxObserDate(extremeMaxInfo.get(station.getStationNo()).get(0).getStartTime() + "-" + extremeMaxInfo.get(station.getStationNo()).get(0).getEndTime());
                    }
                    extremeAnaStationMapDTOS.add(extremeAnaStationDTO);
                }

            }
        });




        //表格数据
        extremeProcInfoMaps.forEach((stationNo,rainInfoAnalysDTOList) -> {
            //循环 rainInfoAnalysDTOList
            rainInfoAnalysDTOList.stream().forEach(x -> {
                ExtremeAnaStationDTO extremeAnaStationDTO = new ExtremeAnaStationDTO();
                extremeAnaStationDTO.setStationNo(stationNo);
                extremeAnaStationDTO.setStationName(stringStationMap.get(stationNo).getCity());
                extremeAnaStationDTO.setProvince(stringStationMap.get(stationNo).getProvince());
                extremeAnaStationDTO.setRiver(stringStationMap.get(stationNo).getRiver());
                extremeAnaStationDTO.setRegion(stringStationMap.get(stationNo).getRegion());
                extremeAnaStationDTO.setLat((Double.valueOf(stringStationMap.get(stationNo).getLat())));
                extremeAnaStationDTO.setLon((Double.valueOf(stringStationMap.get(stationNo).getLon())));
                extremeAnaStationDTO.setYear(x.getYear());
                extremeAnaStationDTO.setObserDate(x.getEndTime());
                extremeAnaStationDTO.setPreObserDate(x.getStartTime());
                Double nowVlaue = x.getValues();
                extremeAnaStationDTO.setValue(nowVlaue);
                //查看阈值
                Double extremeValue = getExtremeProcValue(stationNo, type, flag, extremeInfoMaps);
                extremeAnaStationDTO.setExtremeValue(extremeValue);

                if (CollectionUtils.isEmpty(extremeMaxInfo.get(stationNo))){
                    extremeAnaStationDTO.setMaxObserDate("999999");
                    extremeAnaStationDTO.setMaxValue(999999d);
                    extremeAnaStationDTO.setIsExtreme(false);
                    extremeAnaStationDTO.setIsMaxValue(false);
                }else{
                    Double value = Double.valueOf(extremeMaxInfo.get(stationNo).get(0).getValues());
                    extremeAnaStationDTO.setMaxValue(value);
                    extremeAnaStationDTO.setIsExtreme(x.getValues().compareTo(extremeValue) >= 0);
                    extremeAnaStationDTO.setIsMaxValue(x.getValues().compareTo(value) >= 0);
                    extremeAnaStationDTO.setMaxObserDate(extremeMaxInfo.get(stationNo).get(0).getStartTime() + "-" + extremeMaxInfo.get(stationNo).get(0).getEndTime());
                }
                extremeAnaStationDTOS.add(extremeAnaStationDTO);
            });
        });

        //图数据 站次比
        //处理表格数据 按照年分组 extremeInfoYearMaps
        Map<String, List<ExtremeAnaStationDTO>> extremeInfoYearMaps = extremeAnaStationDTOS.stream().collect(Collectors.groupingBy(ExtremeAnaStationDTO::getYear));

        //todo 查询每一年有多少有效站点
        Map<String, Integer> stringIntegerMap = rainDayService.queryYearValidStNum(riverName);


        //循环所有年份
        List<ExtremeYearDataDTO> extremeYearDataDTOS = new ArrayList<>();
        for (int i = Integer.parseInt(leftTime.substring(0, 4)); i <= Integer.parseInt(rightTime.substring(0, 4)); i++) {
            ExtremeYearDataDTO extremeYearDataDTO = new ExtremeYearDataDTO();
            List<ExtremeAnaStationDTO> extremeAnaStationDTOS1 = extremeInfoYearMaps.get(String.valueOf(i));
            //当年有效站数
            Integer validStNum = stringIntegerMap.get(String.valueOf(i));
            if (!CollectionUtils.isEmpty(extremeAnaStationDTOS1)){
                //    //站次比
                Double rate = new Double(0d);
                //   //极端站次
                int extremeCount = extremeAnaStationDTOS1.size();
                //    //极端站数
                Set<String> stationNos = extremeAnaStationDTOS1.stream().map(ExtremeAnaStationDTO::getStationNo).collect(Collectors.toSet());
                int extremeNum = stationNos.size();
                if (!Objects.equals(validStNum,0)){
                    rate = Double.valueOf(new BigDecimal(extremeCount).divide(new BigDecimal(stringIntegerMap.get(String.valueOf(i))), 4, BigDecimal.ROUND_HALF_UP).doubleValue());
                }
                extremeYearDataDTO.setExtremeCount(extremeCount).setYear(String.valueOf(i)).setExtremeNum(extremeNum).setRate(NumUtil.save2Point(rate * 100));
            }else{
                extremeYearDataDTO.setExtremeCount(0).setYear(String.valueOf(i)).setExtremeNum(0).setRate(0d);
            }
            extremeYearDataDTO.setTotalStations(validStNum);
            extremeYearDataDTOS.add(extremeYearDataDTO);

        }

        extremeAnalysisVO.setExtremeAnaStationMapDTO(extremeAnaStationMapDTOS);
        extremeAnalysisVO.setExtremeAnaStationDTOList(extremeAnaStationDTOS);
        extremeAnalysisVO.setExtremeYearDataDTOS(extremeYearDataDTOS);
        log.info("极端事件 极端过程接口总用时{}",System.currentTimeMillis() - st );
        return ResultVOUtil.success(extremeAnalysisVO);
    }

    private ExtremeAnaStationDTO getExtremeAnaStationDTO(String flag, Integer type, Map<String, List<ProcInfoAnalysDTO>> extremeMaxInfo, Map<String, List<StationExtremeInfo>> extremeInfoMaps, Map<String, Station> stringStationMap, Station station) {
        ExtremeAnaStationDTO extremeAnaStationDTO = new ExtremeAnaStationDTO();
        extremeAnaStationDTO.setStationNo(station.getStationNo());
        extremeAnaStationDTO.setStationName(stringStationMap.get(station.getStationNo()).getCity());
        extremeAnaStationDTO.setProvince(stringStationMap.get(station.getStationNo()).getProvince());
        extremeAnaStationDTO.setRiver(stringStationMap.get(station.getStationNo()).getRiver());
        extremeAnaStationDTO.setRegion(stringStationMap.get(station.getStationNo()).getRegion());
        extremeAnaStationDTO.setLat((Double.valueOf(stringStationMap.get(station.getStationNo()).getLat())));
        extremeAnaStationDTO.setLon((Double.valueOf(stringStationMap.get(station.getStationNo()).getLon())));
        extremeAnaStationDTO.setYear("999999");
        extremeAnaStationDTO.setObserDate("999999");
        extremeAnaStationDTO.setValue(999999d);
        //查看阈值
        //查看阈值
        Double extremeValue = getExtremeProcValue(station.getStationNo(), type, flag, extremeInfoMaps);
        extremeAnaStationDTO.setExtremeValue(extremeValue);

        if (CollectionUtils.isEmpty(extremeMaxInfo.get(station.getStationNo()))) {
            extremeAnaStationDTO.setMaxObserDate("999999");
            extremeAnaStationDTO.setMaxValue(999999d);
            extremeAnaStationDTO.setIsExtreme(false);
            extremeAnaStationDTO.setIsMaxValue(false);
        } else {
            Double value = Double.valueOf(extremeMaxInfo.get(station.getStationNo()).get(0).getValues());
            extremeAnaStationDTO.setMaxValue(value);
            extremeAnaStationDTO.setIsExtreme(false);
            extremeAnaStationDTO.setIsMaxValue(false);
            extremeAnaStationDTO.setMaxObserDate(extremeMaxInfo.get(station.getStationNo()).get(0).getStartTime() + "-" + extremeMaxInfo.get(station.getStationNo()).get(0).getEndTime());
        }
        return extremeAnaStationDTO;
    }

    /**
     * 查询站点的过程的最大值相关信息
     * @param stationNo
     * @param type
     * @param flag
     * @return
     */
    private  Map<String, List<ProcInfoAnalysDTO>>  getExtremeMaxInfo(String stationNo, Integer type, String flag, String leftTime, String rightTime) {
        Long st = System.currentTimeMillis();
        EsAggCondition esAggCondition = EsAggCondition.getInstance();
        Group stationNoGroup = esAggCondition.Group("stationNo");
        stationNoGroup.size(10000);
        Group idGroup = stationNoGroup.groupSub("id").minExtFields("startTime","endTime","year");
        if (Objects.equals(type,1) && Objects.equals(flag,"20")){
            idGroup.min("totalRainFall_20","value");
        }else if (Objects.equals(type,1) && Objects.equals(flag,"08")){
            idGroup.min("totalRainFall_08","value");
        }else if (Objects.equals(type,2) && Objects.equals(flag,"20")){
            idGroup.min("continueDay_20","value");
        }else if (Objects.equals(type,2) && Objects.equals(flag,"08")){
            idGroup.min("continueDay_08","value");
        }else if (Objects.equals(type,3) && Objects.equals(flag,"20")){
            idGroup.min("notContinueDay_20","value");
        }else if (Objects.equals(type,3) && Objects.equals(flag,"08")){
            idGroup.min("notContinueDay_08","value");
        }
        idGroup.order(Order.DESC,"value");
        idGroup.size(1);
        EsCondition condition = esAggCondition.getCondition();
        condition.gte("startTime",leftTime).lte(rightTime);
        if (Objects.equals(type,1) && Objects.equals(flag,"20")){
            condition.eq("totalRainFall_20",null).not();
        }else if (Objects.equals(type,2) && Objects.equals(flag,"20")){
            condition.eq("continueDay_20",null).not();
        }else if (Objects.equals(type,2) && Objects.equals(flag,"08")){
            condition.eq("continueDay_08",null).not();
        }else if (Objects.equals(type,3) && Objects.equals(flag,"20")){
            condition.eq("notContinueDay_20",null).not();
        }else if (Objects.equals(type,3) && Objects.equals(flag,"08")){
            condition.eq("notContinueDay_08",null).not();
        }else if (Objects.equals(type,1) && Objects.equals(flag,"08")){
            condition.eq("totalRainFall_08",null).not();
        }
        if (Objects.nonNull(stationNo)){
            condition.eq("stationNo",stationNo);
        }
        List<ProcInfoAnalysDTO> procInfoAnalysDTOS = extremeProcRepository.selectListAgg(esAggCondition, new DataWrapper<ProcInfoAnalysDTO>() {
            @Override
            public ProcInfoAnalysDTO wrapper(Map<String, Object> data) {
                ProcInfoAnalysDTO procInfoAnalysDTO = new ProcInfoAnalysDTO();
                procInfoAnalysDTO.setStationNo(data.get("stationNo").toString());
                procInfoAnalysDTO.setStartTime(data.get("startTime").toString());
                procInfoAnalysDTO.setEndTime(data.get("endTime").toString());
                if (Objects.equals(type,1)){
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("value").toString())/10);
                }else{
                    procInfoAnalysDTO.setValues(Double.valueOf(data.get("value").toString()));
                }
                procInfoAnalysDTO.setYear(data.get("year").toString());
                return procInfoAnalysDTO;
            }
        });
        log.info("极端事件查询es用时,{}",System.currentTimeMillis() - st);
        Map<String, List<ProcInfoAnalysDTO>> collect = procInfoAnalysDTOS.stream().collect(Collectors.groupingBy(ProcInfoAnalysDTO::getStationNo));
        return collect;
    }



    private Double getExtremeProcValue(String stationNo, Integer type, String flag, Map<String, List<StationExtremeInfo>> extremeInfoMaps) {
        if (CollectionUtils.isEmpty(extremeInfoMaps.get(stationNo))){
            return new Double(999999d);
        }
        if (Objects.equals(type,1)){
            if (Objects.equals(flag,"20")){
                return NumUtil.save2Point(Double.parseDouble(String.valueOf(extremeInfoMaps.get(stationNo).get(0).getContinueThresholdRain())) / 10);
            }
            if (Objects.equals(flag,"08")){
                return NumUtil.save2Point(Double.parseDouble(String.valueOf(extremeInfoMaps.get(stationNo).get(0).getContinueThresholdRain_08())) / 10);
            }
        }
        if (Objects.equals(type,2)){
            if (Objects.equals(flag,"20")){
                return Double.parseDouble(String.valueOf(extremeInfoMaps.get(stationNo).get(0).getContinueThresholdDay()));
            }
            if (Objects.equals(flag,"08")){
                return Double.parseDouble(String.valueOf(extremeInfoMaps.get(stationNo).get(0).getContinueThresholdDay_08()));
            }
        }
        if (Objects.equals(type,3)){
            if (Objects.equals(flag,"20")){
                return Double.parseDouble(String.valueOf(extremeInfoMaps.get(stationNo).get(0).getNotContinueThresholdDay()));
            }
            if (Objects.equals(flag,"08")){
                return Double.parseDouble(String.valueOf(extremeInfoMaps.get(stationNo).get(0).getNotContinueThresholdDay_08()));
            }
        }
        return null;
    }

    /**
     * 初始化极端降水过程数据
     * @param startTime
     * @param endTime
     * @param stationExtremeInfos
     */
    @Override
    public void saveExtremeProcByTime(String startTime, String endTime, List<StationExtremeInfo> stationExtremeInfos) {
        //每个站点每个站点的计算
        Map<String, List<StationExtremeInfo>> extremeInfoMaps = stationExtremeInfos.stream().collect(Collectors.groupingBy(StationExtremeInfo::getStationNo));

        List<Station> stations = StationCache.getStations();


        EsCondition esCondition = EsCondition.getEsCondition();
        esCondition.gte("startTime",startTime).lte(endTime);
        esCondition.size(1000000);
        List<ProcRainInfo> procRainInfos = procRainInfoRepository.selectList(esCondition);
        Map<String, List<ProcRainInfo>> stationProcMaps = procRainInfos.stream().collect(Collectors.groupingBy(ProcRainInfo::getStationNo));
        //循环过滤
        List<ExtremeProcRainInfo> extremeInfoList = new ArrayList<>();
        stationProcMaps.forEach((stationNum,singleStaValues) -> {
            //循环这批数据处理
            singleStaValues.stream().forEach(procRainInfo->{
                if (!CollectionUtils.isEmpty(extremeInfoMaps.get(procRainInfo.getStationNo()))){
                    ExtremeProcRainInfo extremeProcRainInfo = new ExtremeProcRainInfo();
                    //20 降水量
                    if (Objects.nonNull(procRainInfo.getTotalRainFall_20()) && procRainInfo.getTotalRainFall_20().compareTo(extremeInfoMaps.get(procRainInfo.getStationNo()).get(0).getContinueThresholdRain()) >= 0){
                        extremeProcRainInfo.setTotalRainFall_20(procRainInfo.getTotalRainFall_20());
                    }

                    //20 降水日
                    if (Objects.nonNull(procRainInfo.getContinueDay_20()) && procRainInfo.getContinueDay_20().compareTo(extremeInfoMaps.get(procRainInfo.getStationNo()).get(0).getContinueThresholdDay()) >= 0){
                        extremeProcRainInfo.setContinueDay_20(procRainInfo.getContinueDay_20());
                    }

                    //08 降水量
                    if (Objects.nonNull(procRainInfo.getTotalRainFall_08()) && procRainInfo.getTotalRainFall_08().compareTo(extremeInfoMaps.get(procRainInfo.getStationNo()).get(0).getContinueThresholdRain_08()) >= 0){
                        extremeProcRainInfo.setTotalRainFall_08(procRainInfo.getTotalRainFall_20());
                    }

                    //08 降水日
                    if (Objects.nonNull(procRainInfo.getContinueDay_08()) && procRainInfo.getContinueDay_08().compareTo(extremeInfoMaps.get(procRainInfo.getStationNo()).get(0).getContinueThresholdDay_08()) >= 0){
                        extremeProcRainInfo.setContinueDay_08(procRainInfo.getContinueDay_08());
                    }

                    //20 无水日
                    if (Objects.nonNull(procRainInfo.getNotContinueDay_20()) && procRainInfo.getNotContinueDay_20().compareTo(extremeInfoMaps.get(procRainInfo.getStationNo()).get(0).getNotContinueThresholdDay()) >= 0){
                        extremeProcRainInfo.setNotContinueDay_08(procRainInfo.getNotContinueDay_20());
                    }

                    //08 无水日
                    if (Objects.nonNull(procRainInfo.getNotContinueDay_08()) && procRainInfo.getNotContinueDay_08().compareTo(extremeInfoMaps.get(procRainInfo.getStationNo()).get(0).getNotContinueThresholdDay_08()) >= 0){
                        extremeProcRainInfo.setNotContinueDay_08(procRainInfo.getNotContinueDay_08());
                    }

                    // 判断赋值过没
                    if (isNotNull(extremeProcRainInfo)){
                        extremeProcRainInfo.setId(procRainInfo.getId());
                        extremeProcRainInfo.setEndDays(procRainInfo.getEndDays());
                        extremeProcRainInfo.setEndTime(procRainInfo.getEndTime());
                        extremeProcRainInfo.setStartTime(procRainInfo.getStartTime());
                        extremeProcRainInfo.setStationNo(procRainInfo.getStationNo());
                        extremeProcRainInfo.setYear(procRainInfo.getYear());
                        extremeInfoList.add(extremeProcRainInfo);
                    }
                }
            });
        });
        extremeProcRepository.saveBatch(extremeInfoList);
        log.info("保存{}到{}的极端过程数据",startTime,endTime);
    }

    private boolean isNotNull(ExtremeProcRainInfo extremeProcRainInfo){

        return  Objects.nonNull(extremeProcRainInfo.getContinueDay_08()) || Objects.nonNull(extremeProcRainInfo.getContinueDay_20()) ||
                Objects.nonNull(extremeProcRainInfo.getNotContinueDay_08()) || Objects.nonNull(extremeProcRainInfo.getNotContinueDay_20()) ||
                Objects.nonNull(extremeProcRainInfo.getTotalRainFall_08()) || Objects.nonNull(extremeProcRainInfo.getTotalRainFall_20());
    }


}
